package org.apache.flink.runtime.entrypoint.master;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MetricOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.groups.ConsumerManagerMetricGroup;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerGateway;
import org.apache.flink.runtime.resourceconsumer.ConsumingMasterId;
import org.apache.flink.runtime.resourceconsumer.factories.DefaultResourceConsumerResourceMetricGroupFactory;
import org.apache.flink.runtime.resourceconsumer.factories.ConsumerManagerConsumerMetricGroupFactory;
import org.apache.flink.runtime.resourceconsumer.runner.ResourceConsumerRunner;
import org.apache.flink.runtime.resourceconsumer.runner.ResourceConsumerRunnerFactory;
import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.ResourceManagerService;
import org.apache.flink.runtime.resourcemanager.ResourceManagerServiceImpl;
import org.apache.flink.runtime.rest.JobRestEndpointFactory;
import org.apache.flink.runtime.rest.RestEndpointFactory;
import org.apache.flink.runtime.rest.legacy.metrics.MetricFetcher;
import org.apache.flink.runtime.rest.legacy.metrics.MetricFetcherImpl;
import org.apache.flink.runtime.rest.legacy.metrics.VoidMetricFetcher;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.retriever.LeaderGatewayRetriever;
import org.apache.flink.runtime.rpc.retriever.MetricQueryServiceRetriever;
import org.apache.flink.runtime.rpc.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ERException;
import org.apache.flink.util.concurrent.ExponentialBackoffRetryStrategy;
import org.apache.flink.util.concurrent.FutureUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @类名: DefaultMasterComponentFactory
 * @说明: master组件工厂实现
 *
 * @author   leehom
 * @Date	 2023年10月27日 下午1:58:08
 * 修改记录：
 *
 * @see 	 
 * 
 */
public class DefaultMasterComponentFactory implements MasterComponentFactory {

    private final Logger log = LoggerFactory.getLogger(getClass());

    /** 服务组件工厂*/
    @Nonnull private final ResourceConsumerRunnerFactory resourceConsumerRunnerFactory;
    @Nonnull private final ResourceManagerFactory<?> resourceManagerFactory;
    @Nonnull private final RestEndpointFactory<?> restEndpointFactory;

    public DefaultMasterComponentFactory(
            @Nonnull ResourceConsumerRunnerFactory resourceConsumerRunnerFactory,
            @Nonnull ResourceManagerFactory<?> resourceManagerFactory,
            @Nonnull RestEndpointFactory<?> restEndpointFactory) {
        this.resourceConsumerRunnerFactory = resourceConsumerRunnerFactory;
        this.resourceManagerFactory = resourceManagerFactory;
        this.restEndpointFactory = restEndpointFactory;
    }

    // 构建master组件
    @Override
    public MasterComponentOperations create(
    		ConsumerID consumerId,
    		String consumerName,
            Configuration configuration,
            ResourceID resourceId,
            Executor ioExecutor,
            RpcService rpcService,
            HighAvailabilityServices highAvailabilityServices,
            BlobServer blobServer,
            HeartbeatServices heartbeatServices,
            MetricRegistry metricRegistry,
            MetricQueryServiceRetriever metricQueryServiceRetriever,
            FatalErrorHandler fatalErrorHandler)
            throws Exception {
    	// 主节点获取服务
        LeaderRetrievalService resourceConsumerLeaderRetrievalService = null;
        LeaderRetrievalService resourceManagerRetrievalService = null;
        WebMonitorEndpoint<?> webMonitorEndpoint = null;
        // master组件
        ResourceManagerService resourceManagerService = null;
        ResourceConsumerRunner resourceConsumerRunner = null;

        try {
        	// 资源管理器主节点获取服务
            resourceManagerRetrievalService =
                    highAvailabilityServices.getResourceManagerLeaderRetriever();
            // 资源消费者主节点获取服务
            resourceConsumerLeaderRetrievalService =
                    highAvailabilityServices.getResourceConsumerLeaderRetriever();
            
            // 资源管理器网关Retriever
            final LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever =
                    new RpcGatewayRetriever<>(
                            rpcService,
                            ResourceManagerGateway.class,
                            ResourceManagerId::fromUuid,
                            new ExponentialBackoffRetryStrategy(
                                    12, Duration.ofMillis(10), Duration.ofMillis(50)));
            
            // 资源消费者网关Retriever
            final LeaderGatewayRetriever<ResourceConsumerGateway> resourceConsumerGatewayRetriever =
                    new RpcGatewayRetriever<>(
                            rpcService,
                            ResourceConsumerGateway.class,
                            ConsumingMasterId::fromUuid,
                            new ExponentialBackoffRetryStrategy(
                                    12, Duration.ofMillis(10), Duration.ofMillis(50)));
            // 小组件-----
            // 执行器
            final ScheduledExecutorService executor =
                    WebMonitorEndpoint.createExecutorService(
                            configuration.getInteger(RestOptions.SERVER_NUM_THREADS),
                            configuration.getInteger(RestOptions.SERVER_THREAD_PRIORITY),
                            "RestEndpoint");
            
            // rest端-----
            // metrics抓取器
            final long updateInterval =
                    configuration.getLong(MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL);
            final MetricFetcher metricFetcher =
                    updateInterval == 0
                            ? VoidMetricFetcher.INSTANCE
                            : MetricFetcherImpl.fromConfiguration(
                            		null,
                                    configuration,
                                    metricQueryServiceRetriever,
                                    executor);
            // rest端组件
            webMonitorEndpoint =
                    restEndpointFactory.createRestEndpoint(
                    		null, // restful Gateway Retriever，不需要，以后去掉
                            configuration,
                            resourceConsumerGatewayRetriever,
                            resourceManagerGatewayRetriever,
                            blobServer, 
                            executor,
                            metricFetcher,
                            highAvailabilityServices.getClusterRestEndpointLeaderElectionService(),
                            fatalErrorHandler);
            webMonitorEndpoint.start();
            
            final String hostname = RpcUtils.getHostname(rpcService);
            
            // 资源消费者runner
            resourceConsumerRunner = createResourceConsumerRunner(consumerId,
            		consumerName,
                    configuration,
                    rpcService,
                    highAvailabilityServices,
                    heartbeatServices,
                    new DefaultResourceConsumerResourceMetricGroupFactory(
                    		new ConsumerManagerMetricGroup(metricRegistry, 
						            hostname)),
                    fatalErrorHandler);
            resourceConsumerRunner.start();
            
            // 资源管理器运行服务
            resourceManagerService =
                    ResourceManagerServiceImpl.create(
                            resourceManagerFactory,
                            configuration,
                            resourceId,
                            rpcService,
                            highAvailabilityServices,
                            heartbeatServices,
                            fatalErrorHandler,
                            new ClusterInformation(hostname, blobServer.getPort()),
                            webMonitorEndpoint.getRestBaseUrl(),
                            metricRegistry,
                            hostname,
                            ioExecutor);

            log.debug("Starting ResourceManagerService.");
            resourceManagerService.start();
            
            // 启动主节点获取服务
            resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);
            resourceConsumerLeaderRetrievalService.start(resourceConsumerGatewayRetriever);
            // 
            return new MasterComponentHolder(
            			resourceConsumerRunner,
            			resourceConsumerLeaderRetrievalService,
	                    resourceManagerService,
	                    resourceManagerRetrievalService,
	                    webMonitorEndpoint,
	                    fatalErrorHandler);

        } catch (Exception exception) {
            // clean up all started components
            if (resourceConsumerLeaderRetrievalService != null) {
                try {
                    resourceConsumerLeaderRetrievalService.stop();
                } catch (Exception e) {
                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
                }
            }

            if (resourceManagerRetrievalService != null) {
                try {
                    resourceManagerRetrievalService.stop();
                } catch (Exception e) {
                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
                }
            }

            // 服务关闭，future链
            final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);

            if (webMonitorEndpoint != null) {
                terminationFutures.add(webMonitorEndpoint.closeAsync());
            }

            if (resourceManagerService != null) {
                terminationFutures.add(resourceManagerService.closeAsync());
            }

            if (resourceConsumerRunner != null) {
                terminationFutures.add(resourceConsumerRunner.closeAsync());
            }
            
            // 合并终止future
            final FutureUtils.ConjunctFuture<Void> terminationFuture =
                    FutureUtils.completeAll(terminationFutures);
            try {
                terminationFuture.get();
            } catch (Exception e) {
                exception = ExceptionUtils.firstOrSuppressed(e, exception);
            }

            throw new ERException(
                    "Could not create the Master Component.", exception);
        }
    }
    
    private ResourceConsumerRunner createResourceConsumerRunner(
								    		ConsumerID consumerId,
								    		String consumerName,
    										Configuration configuration,
								            RpcService rpcService,
								            HighAvailabilityServices highAvailabilityServices,
								            HeartbeatServices heartbeatServices,
								            ConsumerManagerConsumerMetricGroupFactory consumerManagerConsumerMetricGroupFactory,
								            FatalErrorHandler fatalErrorHandler) throws Exception {
        return resourceConsumerRunnerFactory.createResourceConsumerRunner(
        		consumerId,
        		consumerName,
                configuration,
                rpcService,
                highAvailabilityServices,
                heartbeatServices,
                consumerManagerConsumerMetricGroupFactory,
                fatalErrorHandler,
                System.currentTimeMillis(),
                new DummyBusinessResourceConsumer());
    }

    public static DefaultMasterComponentFactory createMasterComponentFactory(
    		ResourceConsumerRunnerFactory resourceConsumerFactory,
            ResourceManagerFactory<?> resourceManagerFactory) {
        return new DefaultMasterComponentFactory(
        		resourceConsumerFactory,
                resourceManagerFactory,
                JobRestEndpointFactory.INSTANCE);
    }
}
